package com.doeis.log;

import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.doeis.common.utils.DateUtils;
import com.doeis.jpush.DoeisJPush;
import com.doeis.jpush.DoeisJPushWithAlias;
import com.doeis.jpush.JpushBean;
import com.doeis.minaserver.NettyServer;
import com.doeis.minaserver.container.BasicModule;
import com.doeis.minaserver.dao.User;
import com.doeis.minaserver.dao.UserDao;
import com.doeis.minaserver.module.ClientConnectionManager;
import com.doeis.minaserver.module.SocketAcceptorManager;
import com.doeis.minaserver.packet.Packet;
import com.doeis.minaserver.session.ClientConnection;
import com.doeis.uwork.common.Constant;
import com.doeis.uwork.common.PacketUtil;
import com.doeis.uwork.dao.MinaSessionLogMongoDao;
import com.doeis.uwork.dao.OffPacketMongoDao;
import com.doeis.uwork.entity.mongo.GroupMsg;
import com.doeis.uwork.entity.mongo.MethodLog;
import com.doeis.uwork.entity.mongo.MinaActiveSessionLog;
import com.doeis.uwork.entity.mongo.MinaPacketLog;
import com.doeis.uwork.entity.mongo.MinaPushLog;
import com.doeis.uwork.entity.mongo.MinaSessionCountLog;
import com.doeis.uwork.entity.mongo.MinaSessionLog;
import com.doeis.uwork.packet.GroupMsgPacket;
import com.voland.core.db.Entity;
import com.voland.core.utils.IDGenerator;

public class QueueManager extends BasicModule {

	private static final Logger log = Logger.getLogger(QueueManager.class);
	private BlockingQueue logQueue = null;
	private BlockingQueue messageQueue = null;
	private BlockingQueue jpushQueue = null;

	private ConsumerThread consumerThread = null;
	private MsgConsumerThread msgConsumerThread = null;
	private JpushConsumerThread jpushConsumerThread = null;

	private boolean stop = false;

	@Override
	public void initialize(NettyServer server) {
		super.server = server;
	}

	@Override
	public void start() {
		logQueue = new LinkedBlockingQueue();
		messageQueue = new LinkedBlockingQueue();
		jpushQueue = new LinkedBlockingQueue();

		// 开启消费者线程
		log.error("QueueManager 消费者开启 ... ");

		stop = false;
		consumerThread = new ConsumerThread(server.getMinaSessionLogMongoDao());
		consumerThread.start();
		
		msgConsumerThread = new MsgConsumerThread(
				server.getOffPacketMongoDao(), server.getUserDao());
		msgConsumerThread.start();
		
		jpushConsumerThread = new JpushConsumerThread();
		jpushConsumerThread.start();
		
	}

	@Override
	public void stop() {
		super.stop();

		// 开始检测日志是否全部写入到数据库中
		while (!empty() && !emptyMsg() && !emptyJpush()) {
			// 等待队列里面把日志消费完成
		}
		stop = true;
	}

	public QueueManager() {
		super("queue Manager");
		log.error("QueueManager 构造函数开启 ... ");
	}

	// 生产对象
	public synchronized boolean sendLog(ChannelHandlerContext session,
			int status) {
		try {
			MinaSessionLog sessionLog = new MinaSessionLog();
			sessionLog.setStatus(status);
			sessionLog.setId(IDGenerator.uuid());
			sessionLog.setCreated_at(DateUtils.currentDateTime());
			sessionLog.setSessionid(session.channel().id().asShortText());
			sessionLog.setServiceName("");

			AttributeKey<ClientConnection> attributeKey = AttributeKey
					.valueOf(SocketAcceptorManager.CONNECTION);
			Attribute<ClientConnection> attribute = session.attr(attributeKey);
			ClientConnection clienConn = attribute.get();
			if (clienConn != null) {
				sessionLog.setUser_id(clienConn.getUserid());
			}

			// 这个动作比较慢，需要20s
			// SocketAddress socketAddress = session.channel().remoteAddress();
			// if(socketAddress instanceof InetSocketAddress ){
			// InetSocketAddress inSocketAddress = (InetSocketAddress)
			// socketAddress;
			// sessionLog.setRemoteIp(inSocketAddress.getHostName());
			// sessionLog.setRemotePort(inSocketAddress.getPort());
			// }

			SocketAddress socketAddress = session.channel().localAddress();
			if (socketAddress instanceof InetSocketAddress) {
				InetSocketAddress inSocketAddress = (InetSocketAddress) socketAddress;
				sessionLog.setLocalIp(inSocketAddress.getHostName());
				sessionLog.setLocalPort(inSocketAddress.getPort());
			}

			sendLog(sessionLog);
		} catch (Exception e) {
			return false;
		}
		return true;
	}

	// 生产对象
	public synchronized boolean sendLog(Entity entity) {
		try {
			if (entity == null) {
				return true;
			}

			if (logQueue == null) {
				return false;
			}
			logQueue.put(entity);
		} catch (InterruptedException e) {
			e.printStackTrace();
			return false;
		}
		return true;
	}

	// 生产消息对象
	public synchronized boolean sendPacket(Packet packet) {
		try {
			if (packet == null) {
				return true;
			}
			if (messageQueue == null) {
				return false;
			}
			messageQueue.put(packet);
		} catch (InterruptedException e) {
			e.printStackTrace();
			return false;
		}
		return true;
	}

	/**
	 * 方法介绍  生产极光推送信息
	 * @param userid
	 * @param context
	 * @return
	 * @author tianbx
	 * @version 2016年5月21日下午2:54:51
	 */
	public synchronized boolean sendJpush(JpushBean jpushBean) {
		try {
			if (jpushBean == null) {
				return true;
			}

			if (jpushQueue == null) {
				return false;
			}
			jpushQueue.put(jpushBean);
		} catch (InterruptedException e) {
			e.printStackTrace();
			return false;
		}
		return true;
	}

	/**
	 * 方法介绍  判断日志消费是否是空
	 * @return
	 * @author tianbx
	 * @version 2016年5月21日下午2:55:27
	 */
	public boolean empty() {
		if (logQueue == null) {
			return true;
		} else {
			return logQueue.isEmpty();
		}
	}

	public boolean emptyMsg() {
		if (messageQueue == null) {
			return true;
		} else {
			return messageQueue.isEmpty();
		}
	}

	/**
	 * 方法介绍   极光推送消费者是否是空
	 * @return
	 * @author tianbx
	 * @version 2016年5月21日下午2:55:45
	 */
	public boolean emptyJpush() {
		if (jpushQueue == null) {
			return true;
		} else {
			return jpushQueue.isEmpty();
		}
	}

	/***
	 * 类介绍      日志消费者
	 * @author tianbx
	 * @version 2016年5月21日下午2:56:23
	 */
	class ConsumerThread extends Thread {

		MinaSessionLogMongoDao logMongoDao = null;

		public ConsumerThread(MinaSessionLogMongoDao logMongoDao) {
			log.error("QueueManager Consumer构造函数创建 ... ");
			this.logMongoDao = logMongoDao;
		}

		@Override
		public void run() {
			Entity entity = null;
			while (!stop) {
				if (logQueue.isEmpty()) {
					try {
						sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					continue;
				} else {
					try {
						entity = (Entity) logQueue.poll();

						if (entity != null) {
							if (entity instanceof MinaSessionLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MinaSessionLog) entity);
							} else if (entity instanceof MinaSessionCountLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MinaSessionCountLog) entity);
							} else if (entity instanceof MinaActiveSessionLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MinaActiveSessionLog) entity);
							} else if (entity instanceof MinaPacketLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MinaPacketLog) entity);
							} else if (entity instanceof MinaPushLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MinaPushLog) entity);
							} else if (entity instanceof MethodLog) {
								logMongoDao.insert(
										Constant.DATABASE_MONGO_DOEIS,
										(MethodLog) entity);
							}
						}
					} catch (Exception e) {
						log.error("消费session日志报错", e);
					}
				}
			}
		}
	}

	// 消费者
	class MsgConsumerThread extends Thread {

		OffPacketMongoDao offPacketMongoDao = null;
		UserDao userDao = null;

		public MsgConsumerThread(OffPacketMongoDao offPacketMongoDao,
				UserDao userDao) {
			log.error("QueueManager MsgConsumer构造函数创建 ... ");
			this.offPacketMongoDao = offPacketMongoDao;
			this.userDao = userDao;
		}

		@Override
		public void run() {
			Packet packet = null;
			ClientConnectionManager clientConnectionManager = server
					.getClientConnectionManager();
			while (!stop) {
				if (messageQueue.isEmpty()) {
					try {
						sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					continue;
				}

				try {
					packet = (Packet) messageQueue.poll();
					if (packet != null && clientConnectionManager != null) {

						Map<String, ClientConnection> map = clientConnectionManager
								.getAuthConnections();
						ClientConnection clientConnection = map.get(packet.getTo());

						if (packet instanceof GroupMsgPacket) {
							if (clientConnection != null) {
								clientConnection.deliver(packet);
							} else {
								GroupMsg groupmsg = JSON.parseObject(
										packet.getBody(), GroupMsg.class);
								if (groupmsg != null) {
									User fromUser = JSON.parseObject(
											groupmsg.getFromUser(), User.class);
									if (fromUser != null) {
										// 极光推送，只有苹果有
										String content = groupmsg.getContent();
										if ("2".equals(groupmsg.getMsgType())) {
											content = "[图片]";
										}
										DoeisJPush.sendNotification(
												packet.getTo(),
												fromUser.getName() + ":"
														+ content);
									}
								}
							}
							GroupMsg groupMsg = JSON.parseObject(
									packet.getBody(), GroupMsg.class);
							User user = userDao.findUserByid(groupMsg
									.getUserId());
							offPacketMongoDao.insert(
									user.getSwitchboard_id(),
									PacketUtil.cloneOffPacket(packet,
											user.getId(), groupMsg.getId()));
						}
					}
				} catch (Exception e) {
					log.error("消费msg报错", e);
				}

			}
		}
	}

	/***
	 * 类介绍      极光推送信息消费者
	 * @author tianbx
	 * @version 2016年5月21日下午2:56:45
	 */
	class JpushConsumerThread extends Thread {

		public JpushConsumerThread() {
			log.error("QueueManager JpushConsumer构造函数创建 ... ");
		}

		@Override
		public void run() {
			JpushBean jpushBean = null;
			while (!stop) {
				if (jpushQueue.isEmpty()) {
					try {
						sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					continue;
				}

				try {
					jpushBean = (JpushBean) jpushQueue.poll();
					if (jpushBean != null) {
						if(jpushBean.getUserid()!=null){
							DoeisJPush.sendNotification(jpushBean.getUserid(),
									jpushBean.getContent());
						}
						
						if(jpushBean.getUseridSet()!=null){
							DoeisJPushWithAlias.sendNotificationSet(jpushBean.getUseridSet(),
									jpushBean.getContent());
						}
					}
				} catch (Exception e) {
					log.error("消费msg报错", e);
				}

			}
		}
	}

}